package com.shujia.mr.kqzl2;

import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

import java.io.IOException;
import java.util.ArrayList;

public class KqzlReducer2 extends Reducer<Text, Text, Text, Text> {
    @Override
    protected void reduce(Text key, Iterable<Text> values, Reducer<Text, Text, Text, Text>.Context context) throws IOException, InterruptedException {
        //<"1279", ["$20180103-241","$20180104-217","#合肥-高新区","$20180101-128","$20180102-24"]>
        //创建一个集合，存储分组后该检测到每一天的平均pm2.5的值
        ArrayList<String> pmList = new ArrayList<>();
        //创建一个遍历存储该检测点的地点名称
        String idName = null;

        //遍历迭代器
        for (Text value : values) {
            String info = value.toString();
            if(info.startsWith("#")){
                idName = info.substring(1).replace("-","\t");
            }else if(info.startsWith("$")){
                pmList.add(info.substring(1));
            }
        }

        //遍历集合将遍历值拼接输出
        for (String pmInfo : pmList) {
            String dateTime = pmInfo.split("-")[0];
            String pm = pmInfo.split("-")[1];
            // 20180101 安徽省 合肥市 128
            context.write(new Text(dateTime+"\t"+idName),new Text(pm));
        }

    }
}
